package cn.z2huo.demo.kafka.spring.producer.config;

import cn.z2huo.demo.kafka.spring.producer.properties.TestTopicProperties;
import cn.z2huo.demo.kafka.spring.producer.support.TestTopicKafkaSendResultHandler;
import lombok.RequiredArgsConstructor;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

import java.util.Map;

/**
 * <p>
 *
 * @author z2huo
 */
@RequiredArgsConstructor
@Configuration
@EnableConfigurationProperties(TestTopicProperties.class)
public class TestTopicProducerConfig {

    private final TestTopicProperties testTopicProperties;

    private final TestTopicKafkaSendResultHandler testTopicKafkaSendResultHandler;

    @Bean(name = "testTopicKafkaProducerFactory")
    public ProducerFactory<String, String> kafkaProducerFactory() {
        DefaultKafkaProducerFactory<String, String> factory =
                new DefaultKafkaProducerFactory<>(testTopicProperties.buildProducerProperties());
//        customizers.orderedStream().forEach((customizer) -> customizer.customize(factory));
        return factory;
    }

    @Primary
    @Bean(name = "testTopicKafkaTemplate")
    public KafkaTemplate<String, String> testTopicKafkaTemplate(
            @Qualifier("testTopicKafkaProducerFactory") ProducerFactory<String, String> producerFactory) {
        KafkaTemplate<String, String> kafkaTemplate = new KafkaTemplate<>(producerFactory);
        kafkaTemplate.setProducerListener(testTopicKafkaSendResultHandler);
        return kafkaTemplate;
    }

    @Bean(name = "testTopicCustomConfigKafkaTemplate")
    public KafkaTemplate<String, String> testTopicCustomConfigKafkaTemplate(
            @Qualifier("testTopicKafkaProducerFactory") ProducerFactory<String, String> producerFactory) {
        KafkaTemplate<String, String> kafkaTemplate = new KafkaTemplate<>(producerFactory,
                Map.of(
                        ProducerConfig.BATCH_SIZE_CONFIG, 30000,
                        ProducerConfig.LINGER_MS_CONFIG, 10000
                ));
        return kafkaTemplate;
    }


}
